package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/segmentio/kafka-go"
)

// Kafka 配置
const (
	kafkaAddress = "127.0.0.1:9092"
	topicName    = "test"
)

// 全局 Kafka Writer 实例
var writer *kafka.Writer

func initKafkaWriter() {
	writer = &kafka.Writer{
		Addr:         kafka.TCP(kafkaAddress), // 设置Kafka的TCP连接地址（配置）
		Topic:        topicName,               // 设置要写入的主题名称（配置）
		Balancer:     &kafka.LeastBytes{},     // 选择分区策略，这里使用最小字节策略（保持）
		BatchSize:    100,                     // 设置批次大小，以消息数量为单位（选填）
		BatchBytes:   1024 * 1024,             // 设置批次字节大小上限（选填）
		BatchTimeout: 1 * time.Second,         // 批次超时时间，触发批量发送的超时机制（选填）
		Async:        true,                    // 使用异步模式发送消息，提高吞吐量（保持）
		RequiredAcks: kafka.RequireOne,        // 设置应答级别，仅需一个副本确认，平衡可靠性和性能（保持，默认kafka.RequireNone）
		Compression:  kafka.Snappy,            // 使用Snappy压缩以减少网络传输量（选填）
	}

	// 测试连接 Kafka
	conn, err := kafka.Dial("tcp", kafkaAddress)
	if err != nil {
		fmt.Printf("连接Kafka broker失败: %v", err)
		return
	}
	defer conn.Close()

	// 确保 topic 存在
	_, err = conn.ReadPartitions(topicName)
	if err != nil {
		fmt.Printf("找不到topic %s: %v", topicName, err)
		writer.Close()
		return
	}
	fmt.Println("连接Kafka broker成功")
}

func main() {
	initKafkaWriter()
	r := gin.Default()
	r.GET("/write", func(c *gin.Context) {
		// 写入消息
		err := writer.WriteMessages(context.Background(),
			kafka.Message{
				Value: []byte("Hello Kafka"),
			},
		)
		if err != nil {
			log.Printf("写入消息失败: %v", err)
			c.JSON(500, gin.H{"error": "写入消息失败"})
			return
		}

		log.Println("写入消息成功")
		c.JSON(200, gin.H{"message": "写入消息成功"})
	})

	// 启动服务器
	r.Run()

}
